package main

import (
	"encoding/json"
	"fmt"
	amqp "github.com/rabbitmq/amqp091-go"
	"time"
)

type Obj struct {
	Item1 string `json:"item1"`
	Item2 string `json:"item2"`
	Item3 string `json:"item3"`
}

func StartAMQPConsume() {
	defer func() {
		if err := recover(); err != nil {
			time.Sleep(3 * time.Second)
			fmt.Println("休息3秒")
			StartAMQPConsume()
		}
	}()

	conn, err := amqp.Dial("amqp://user:password@182.92.212.121:5672/")
	if err != nil {
		fmt.Println(err)
	}
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		fmt.Println(err)
	}
	defer ch.Close()

	closeChan := make(chan *amqp.Error, 1)
	// 一旦消费者的channel有错误，产生一个amqp.Error，channel监听并捕捉到这个错误
	notifyClose := ch.NotifyClose(closeChan)
	closeFlag := false

	msgs, err := ch.Consume(
		"test-queue", // 队列名
		"",           // 消费者名，用来区分多个消费者，以实现公平分发或均等分发策略
		true,         // 是否自动应答
		false,        // 是否排他
		false,        // 是否接收只同一个连接中的消息，若为true，则只能接收别的conn中发送的消息
		true,         // 队列消费是否阻塞
		nil)

	var obj Obj

	for {
		select {
		case e := <-notifyClose:
			fmt.Printf("chan通道错误，e:%s", e.Error())
			close(closeChan)
			time.Sleep(3 * time.Second)
			StartAMQPConsume()
			closeFlag = true
		case msg := <-msgs:
			if err := json.Unmarshal(msg.Body, &obj); err != nil {
				fmt.Println(err.Error())
			}
			fmt.Println(obj.Item1)
		}

		if closeFlag {
			break
		}
	}
}

func main() {
	StartAMQPConsume()
}
